#include <ros/ros.h>
#include "common_lib/socket_session.h"

using namespace Socket;

SocketSession::SocketSession(boost::shared_ptr<boost::asio::io_service> ios, boost::asio::ip::tcp::resolver::results_type endpoints, 
                             SocketInterface *client) noexcept
: io_service_{ios},
  socket_(*io_service_),
  deadline_(*io_service_),
  heartbeat_timer_(*io_service_),
  endpoints_(endpoints)
{
    read_callback_ = boost::bind(&SocketInterface::ReadCallback, client, &input_buffer_, &error_code_);
    write_callback_ = boost::bind(&SocketInterface::HandleWrite, client, &write_length_, &error_code_);
    client_connect_callback_ = boost::bind(&SocketInterface::OnConnectStatus, client, &client_is_connect_, &error_code_);

    deadline_.async_wait(boost::bind(&SocketSession::CheckDeadline, this));
}

SocketSession::SocketSession(boost::shared_ptr<boost::asio::io_service> ios, SocketInterface *server) noexcept
: io_service_{ios},
  socket_{*io_service_},
  heartbeat_timer_(*io_service_),
  deadline_(*io_service_)
{
    read_callback_ = boost::bind(&SocketInterface::ReadCallback, server, &input_buffer_, &error_code_);
    write_callback_ = boost::bind(&SocketInterface::HandleWrite, server, &write_length_, &error_code_);

    deadline_.async_wait(boost::bind(&SocketSession::CheckDeadline, this));
}

SocketSession::SocketSession(boost::shared_ptr<boost::asio::io_service> ios) noexcept
: io_service_{ios},
  socket_{*io_service_},
  deadline_(*io_service_),
  heartbeat_timer_(*io_service_)
{
    deadline_.async_wait(boost::bind(&SocketSession::CheckDeadline, this));
}

SocketSession::~SocketSession() noexcept
{
    if (socket_.is_open())
    {
        printf("~SocketSession() --> close\n");
        socket_.shutdown(boost::asio::socket_base::shutdown_both);
        socket_.close();
    }
    read_callback_.clear();
    write_callback_.clear();
    client_connect_callback_.clear();

    executors_thread_group_.interrupt_all();
    executors_thread_group_.join_all();
}

bool SocketSession::Start() noexcept
{
    deadline_.expires_after(std::chrono::seconds(100));

    boost::thread *read_thread = executors_thread_group_.create_thread(boost::bind(&SocketSession::StartRead, this));
    std::cout << read_thread->get_id() << std::endl;
    
    boost::thread *write_thread = executors_thread_group_.create_thread(boost::bind(&SocketSession::StartWrite, this));
    std::cout << write_thread->get_id() << std::endl;
    
    return true;
}

void SocketSession::Stop() noexcept
{
    printf("SocketSession::Stop()\n");
    boost::system::error_code stopped_error;
    socket_.close(stopped_error);
    deadline_.cancel();
    heartbeat_timer_.cancel();

    executors_thread_group_.interrupt_all();
    executors_thread_group_.join_all();
}

void SocketSession::StartConnect(boost::asio::ip::tcp::resolver::results_type::iterator endpoint_iter)
{
    if (endpoint_iter != endpoints_.end())
    {
        std::cout << "Trying " << endpoint_iter->endpoint() << "...\n";
        
        // Set a deadline for the connect operation.
        deadline_.expires_after(std::chrono::seconds(10));

        // Start the asynchronous connect operation.
        socket_.async_connect(endpoint_iter->endpoint(), boost::bind(&SocketSession::HandleConnect, this, _1, endpoint_iter));
    }
    else
    {
        Stop();
    }
}

void SocketSession::HandleConnect(const boost::system::error_code &error, boost::asio::ip::tcp::resolver::results_type::iterator endpoint_iter)
{
    client_is_connect_ = false;
    boost::asio::ip::tcp::resolver::results_type::iterator tmp = endpoint_iter;
    tmp ++;
    printf("HandleConnect\n");
    auto call = [=]()
    {
        if (!client_connect_callback_.empty())
        {
            client_connect_callback_();
        }
    };

    if (!socket_.is_open())
    {
        std::cout << "Connect timed out" << std::endl;
        if (tmp == endpoints_.end())
            call();
        StartConnect(++endpoint_iter);
    }
    else if (error)
    {
        error_code_ = error;
        std::cout << "Connect error: " << error.message() << "\n";

        // We need to close the socket used in the previous connection attempt before starting a new one.
        socket_.close();

        if (tmp == endpoints_.end())
            call();

        // Try the next available endpoint.
        StartConnect(++endpoint_iter);
    }
    else // Otherwise we have successfully established a connection.
    {
        std::cout << "Connected to " << endpoint_iter->endpoint() << "\n";
        client_is_connect_ = true;
        call();
#if 1
        boost::thread *read_thread = executors_thread_group_.create_thread(boost::bind(&SocketSession::StartRead, this));
        std::cout << read_thread->get_id() << std::endl;
        
        boost::thread *write_thread = executors_thread_group_.create_thread(boost::bind(&SocketSession::StartWrite, this));
        std::cout << write_thread->get_id() << std::endl;
#else
        executors_thread_group_.create_thread(boost::bind(&SocketSession::ThreadFun1, this, std::string("线程1")));
        executors_thread_group_.create_thread(boost::bind(&SocketSession::ThreadFun2, this, std::string("线程2")));
#endif
    }
}

void SocketSession::StartRead()
{
    deadline_.expires_after(std::chrono::seconds(10));

    input_buffer_.clear();
    memset(buffer_, 0x00, sizeof(buffer_));
    socket_.async_read_some(boost::asio::buffer(buffer_, sizeof(buffer_)), boost::bind(&SocketSession::HandleRead, this, _1, _2));
}

void SocketSession::HandleRead(const boost::system::error_code& error, std::size_t length)
{
    if (stopped_)
        return;

    if (!this->read_callback_.empty())
    {
        error_code_ = error;
        input_buffer_ = buffer_;
        this->read_callback_();
    }
    ROS_INFO("HandleRead() --> size: %ld, error: %s", length, error.message().c_str());
    if (!error)
    {
        
        ROS_INFO("Read Buffer: %s", input_buffer_.c_str());
        StartRead();
    }
    else
    {
        std::cout << "Error on receive: " << error.message() << ", code: " << error.value() << "\n";
        Stop();
    }
}

void SocketSession::CheckDeadline()
{
    if (stopped_)
        return;

    // Check whether the deadline has passed. We compare the deadline against
    // the current time since a new asynchronous operation may have moved the
    // deadline before this actor had a chance to run.
    if (deadline_.expiry() <= boost::asio::steady_timer::clock_type::now())
    {
        // The deadline has passed. The socket is closed so that any outstanding
        // asynchronous operations are cancelled.
        printf("CheckDeadline() --> close\n");
        socket_.close();

        // There is no longer an active deadline. The expiry is set to the
        // maximum time point so that the actor takes no action until a new
        // deadline is set.
        deadline_.expires_at(boost::asio::steady_timer::time_point::max());
    }

    deadline_.async_wait(boost::bind(&SocketSession::CheckDeadline, this));
}

bool SocketSession::Write(const std::string &str)
{
    boost::asio::async_write(socket_, boost::asio::buffer(str, str.length()), [this](const boost::system::error_code& error, std::size_t length)
    {
        if (!this->write_callback_.empty())
        {
            this->error_code_ = error;
            this->write_length_ = length;
            this->write_callback_();
        }
        if (error)
        {
            std::cout << "Error on write: " << error.message() << ", code: " << error.value() << "\n";
            this->Stop();
            return false;
        }
        return true;
    });
    return false;
}

void SocketSession::StartWrite()
{
    boost::this_thread::sleep(boost::posix_time::millisec(30));
    std::string str("123#456#789#");
    str += std::to_string(get_peer_port(socket_));

    ROS_INFO("Session Write %s", str.c_str());
    deadline_.expires_after(std::chrono::seconds(10));
    boost::asio::async_write(socket_, boost::asio::buffer(str, str.length()), boost::bind(&SocketSession::HandleWrite, this, _1, _2));
}

void SocketSession::HandleWrite(const boost::system::error_code& error, std::size_t length)
{
    if (!this->write_callback_.empty())
    {
        this->error_code_ = error;
        this->write_length_ = length;
        this->write_callback_();
    }
    if (error)
    {
        printf("error: %s\n", error.message().c_str());
        Stop();
    }
    else
    {

        heartbeat_timer_.expires_after(std::chrono::milliseconds(100));
        heartbeat_timer_.async_wait(boost::bind(&SocketSession::StartWrite, this));
    }
}

std::string SocketSession::get_peer_ip(const boost::asio::ip::tcp::socket &socket) noexcept
{
    return socket.remote_endpoint().address().to_string();
}

unsigned short SocketSession::get_peer_port(const boost::asio::ip::tcp::socket &socket) noexcept
{
    return socket.remote_endpoint().port();
}